package com.gyg.threadPool.customThreadPool;

import lombok.extern.slf4j.Slf4j;

import java.util.ArrayDeque;
import java.util.Deque;
import java.util.HashSet;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

/**
 * 自定义线程池
 */
@Slf4j
public class CustomPoolTest {
    public static void main(String[] args) {
        //创建线程池
        ThreadPool threadPool = new ThreadPool(2, 1000, TimeUnit.MILLISECONDS, 1, (queue, task) -> {
            //taskQueue.put(task);
            //线程等待的策略
            //（1）死等
            //queue.put(task);
            //（2）带超时等待
            //queue.put(task,1,TimeUnit.SECONDS);
            //（3）让调用者放弃任务执行
            //log.info("放弃执行...{}", task);
            //（4）让调用者抛出异常
            //throw new RuntimeException("任务执行失败" + task);
            //（5）让调用者自己执行任务
            task.run();
        });
        for (int i = 0; i < 4; i++) {
            int temp = i;
            threadPool.execute(() -> {
                try {
                    Thread.sleep(10000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                log.info("T-{}", temp);
            });
        }
    }
}

@FunctionalInterface
interface RejectPolicy<T> {
    void reject(BlockingQueue<T> queue, T task);
}


/**
 * 线程池
 */
@Slf4j
class ThreadPool {
    /**
     * 任务队列
     */
    private BlockingQueue<Runnable> taskQueue;
    /**
     * 线程集合
     */
    private HashSet<Worker> workerSet = new HashSet<>();
    /**
     * 核心线程数
     */
    private int coreSize;
    /**
     * 获取任务的超时时间
     */
    private long timeOut;
    /**
     * 超时时间单位
     */
    private TimeUnit timeUnit;
    /**
     * 拒绝策略
     */
    private RejectPolicy<Runnable> rejectPolicy;

    /**
     * 执行方法
     */
    public void execute(Runnable task) {
        synchronized (workerSet) {
            //1、线程集合中的线程小于核心线程数，创建线程
            if (workerSet.size() < coreSize) {
                //创建一个线程并将任务提交给线程执行
                Worker worker = new Worker(task);
                //把线程加入到线程队列
                log.info("新增Worker：{}到线程集合，任务：{}", worker, task);
                workerSet.add(worker);
                //运行线程
                worker.start();
            } else {
                //2、线程集合中的线程大于核心线程数，加入阻塞队列
                //taskQueue.put(task);
                //线程等待的策略
                //（1）死等
                //（2）带超时等待
                //（3）让调用者放弃任务执行
                //（4）让调用者抛出异常
                //（5）让调用者自己执行任务
                taskQueue.tryPut(task, rejectPolicy);

            }
        }
    }


    public ThreadPool(int coreSize, long timeOut, TimeUnit timeUnit, int capcity, RejectPolicy<Runnable> rejectPolicy) {
        this.coreSize = coreSize;
        this.timeOut = timeOut;
        this.timeUnit = timeUnit;
        this.taskQueue = new BlockingQueue<>(capcity);
        this.rejectPolicy = rejectPolicy;
    }

    /**
     * 线程封装类
     */
    class Worker extends Thread {
        private Runnable task;

        public Worker(Runnable task) {
            this.task = task;
        }

        @Override
        public void run() {
            //执行任务
            //1、task任务不为为空时执行
            //2、task任务执行完毕，并且阻塞队列中存在未执行的任务
            //3、以上两项都不满足，跳出while
            //TODO:获取队列中任务的两种方式：1、未带超时时间、2、带超时时间
            while (task != null || (task = taskQueue.pull(timeOut, timeUnit)) != null) {
                try {
                    log.info("正在执行...：{}", task);
                    task.run();
                } catch (Exception e) {
                    e.printStackTrace();
                } finally {
                    //任务运行完之后要置为空
                    task = null;
                }
            }
            //该线程没有任务可以执行，将线程从队列中移除
            synchronized (workerSet) {
                log.info("移除线程：", this);
                workerSet.remove(this);
            }

        }
    }
}

/**
 * 阻塞队列
 *
 * @param <T>
 */
@Slf4j
class BlockingQueue<T> {
    //1、任务队列
    private Deque<T> deque = new ArrayDeque<>();
    //2、锁
    private ReentrantLock lock = new ReentrantLock();
    //3、生产者条件变量
    private Condition fullWaitSet = lock.newCondition();
    //4、消费者条件变量
    private Condition emptyWaitSet = lock.newCondition();
    //5、队列容量
    private int capcity;

    public BlockingQueue(int capcity) {
        this.capcity = capcity;
    }

    /**
     * 带超时时间的阻塞获取
     *
     * @param timeout
     * @param unit
     * @return
     */
    public T pull(long timeout, TimeUnit unit) {
        //方法加锁
        lock.lock();
        try {
            //时间统一转换为纳秒
            long nanos = unit.toNanos(timeout);
            while (deque.isEmpty()) {
                try {
                    if (nanos <= 0) {
                        return null;
                    }
                    //防止虚假唤醒问题，即没等够指定时间就被唤醒
                    nanos = emptyWaitSet.awaitNanos(nanos);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            //获取到队列第一个元素
            T t = deque.removeFirst();
            //放开阻塞的变量
            fullWaitSet.signal();
            return t;
        } finally {
            lock.unlock();
        }
    }

    /**
     * 不带超时时间的阻塞获取
     *
     * @return
     */
    public T take() {
        //方法加锁
        lock.lock();
        try {
            //时间统一转换为纳秒
            while (deque.isEmpty()) {
                try {
                    //阻塞
                    emptyWaitSet.await();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            //获取到队列第一个元素
            T t = deque.removeFirst();
            //放开阻塞的变量
            fullWaitSet.signal();
            return t;
        } finally {
            lock.unlock();
        }
    }

    /**
     * 向队列中提交元素
     *
     * @param task
     */
    public void put(T task) {
        lock.lock();
        try {
            while (deque.size() == capcity) {
                try {
                    //生产者阻塞
                    log.info("任务等待加入阻塞队列：{}......", task);
                    fullWaitSet.await();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            //在队列末尾添加元素
            log.info("任务加入阻塞队列：{}", task);
            deque.addLast(task);
            //唤醒消费者变量
            emptyWaitSet.signal();
        } finally {
            lock.unlock();
        }
    }

    /**
     * 带超时时间的阻塞添加方法
     */
    public boolean put(T task, long timeOut, TimeUnit timeUnit) {
        lock.lock();
        try {
            long nanos = timeUnit.toNanos(timeOut);
            while (deque.size() == capcity) {
                try {
                    //生产者阻塞
                    log.info("任务等待加入阻塞队列：{}......", task);
                    //到达超时时间时未能加入队列，
                    if (nanos <= 0) {
                        return false;
                    }
                    //带超时时间的阻塞
                    nanos = fullWaitSet.awaitNanos(nanos);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            //在队列末尾添加元素
            log.info("任务加入阻塞队列：{}", task);
            deque.addLast(task);
            //唤醒消费者变量
            emptyWaitSet.signal();
            return true;
        } finally {
            lock.unlock();
        }
    }

    /**
     * 获取队列长度
     *
     * @return
     */
    public int size() {
        return deque.size();
    }

    /**
     * 带拒绝策略的提交
     *
     * @param task
     * @param rejectPolicy
     */
    public void tryPut(T task, RejectPolicy<T> rejectPolicy) {
        lock.lock();
        try {
            if (deque.size() == capcity) {
                rejectPolicy.reject(this, task);
            } else {
                log.info("加入任务队列{}", task);
                deque.addLast(task);
                emptyWaitSet.signal();
            }
        } finally {
            lock.unlock();
        }
    }
}


